﻿using Microsoft.ServiceBus.Messaging;
using Sustainalytics.Utils;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace Sustainalytics.Replication.Publisher
{
    public class ServiceBusPublisher
    {
        TopicClient topicClient;
        string regionEndpoint;
        string sessionId;
        Queue<BrokeredMessage> queue;

        public ServiceBusPublisher(Dictionary<string, string> serviceBusParams)
        {
            try
            {
                if (ValidateParams(serviceBusParams))
                {
                    regionEndpoint = serviceBusParams["RegionEndpoint"];
                    sessionId = serviceBusParams["SessionId"];
                    topicClient = TopicClient.CreateFromConnectionString(serviceBusParams["ConnectionString"], serviceBusParams["TopicPath"]);
                    queue = new Queue<BrokeredMessage>();
                    ProcessQueue();
                }
            }
            catch (Exception ex)
            {
                CommonEventSource.Log.Exception(ex);
            }
        }

        private bool ValidateParams(Dictionary<string, string> serviceBusParams)
        {
            bool result = true;

            List<string> keys = new List<string>() { "RegionEndpoint", "SessionId", "ConnectionString", "TopicPath" };

            foreach (string key in keys)
            {
                if (!serviceBusParams.ContainsKey(key) || string.IsNullOrEmpty(serviceBusParams[key]))
                    result = false;                
            }

            return result;
        }

        void ProcessQueue()
        {
            Task.Run(async () =>
            {
                while (true)
                {
                    while (queue.Count > 0)
                    {
                        try
                        {
                            var brokeredMessage = queue.Dequeue();
                            await topicClient.SendAsync(brokeredMessage);
                        }
                        catch (Exception ex)
                        {
                            CommonEventSource.Log.Exception(ex);
                        }
                    }
                    await Task.Delay(200);
                }
            });
        }

        /// <summary>
        /// Add message to the Topic
        /// </summary>
        /// <param name="message">Entity object JSON serialized</param>
        /// <param name="collection">Entity/Collection name</param>
        /// <param name="operation">CRUD Operation</param>
        public void SendMessage(string message, ReplicationUtils.EntityType entityType, ReplicationUtils.Operation operation)
        {
            Task.Run(() => SendMessage(message, entityType, operation, ReplicationUtils.ConnectionStringType.ClientPlatform)).ConfigureAwait(false);
        }

        /// <summary>
        /// Add message to the Topic
        /// </summary>
        /// <param name="message">Entity object JSON serialized</param>
        /// <param name="collection">Entity/Collection name</param>
        /// <param name="operation">CRUD Operation</param>
        public void SendMessage(string message, ReplicationUtils.EntityType entityType, ReplicationUtils.Operation operation, ReplicationUtils.ConnectionStringType connectionStringType)
        {
            try
            {
                BrokeredMessage brokeredMessage = new BrokeredMessage(message);
                brokeredMessage.Properties["Region"] = regionEndpoint;
                brokeredMessage.Properties["EntityType"] = Enum.GetName(typeof(ReplicationUtils.EntityType), entityType);
                brokeredMessage.Properties["Operation"] = Enum.GetName(typeof(ReplicationUtils.Operation), operation);
                brokeredMessage.Properties["ConnectionStringType"] = Enum.GetName(typeof(ReplicationUtils.ConnectionStringType), connectionStringType);
                brokeredMessage.SessionId = sessionId;

                queue.Enqueue(brokeredMessage);
            }
            catch (Exception ex)
            {
                CommonEventSource.Log.Exception(ex);
            }
        }
    }
}
